//
//

#include "StreamPuller.h"
#include "Log.h"
#include "StringHelp.h"

#ifdef __cplusplus
extern "C" {
#endif
#include <libavcodec/avcodec.h>
#include <libavformat/avformat.h>
#include <libavutil/avutil.h>
#include <libavutil/imgutils.h>
#include <libavutil/pixdesc.h>
#include <libavutil/time.h>
#include <libswscale/swscale.h>
#ifdef __cplusplus
}
#endif


#define SLOG(level) LOG(level) << "[taskId:" << mTaskId << "]"

StreamPuller::StreamPuller(TASK_ID_TYPE taskId, std::string url, int fps, std::string serverAddress, int retry_times)
    : mTaskId(std::move(taskId)), mURL(std::move(url)), mFps(fps), 
    mServerAddress(std::move(serverAddress)), mOnError(nullptr), retryTimes(retry_times){
    mTaskStatus = ThreadStatus::STOP;
}

void StreamPuller::Stop() {
    mStop = true;
    SLOG(DEBUG) << "Task Stopping...";
    if (mStreamThread && (std::this_thread::get_id() != mStreamThread->get_id())) {
        SLOG(DEBUG) << "Task Stopping, Stop Pull Thread.";
        try {
            mStreamThread->join();
            mStreamThread = nullptr;
        } catch (std::exception &e) {
            SLOG(ERROR) << "mStreamThread join exception: " << e.what();
        }
    }
    mTaskStatus = ThreadStatus::STOP;
    SLOG(INFO) << "Task Stopped.";
}

void StreamPuller::Start() {
    SLOG(INFO) << "Starting......";
    if (mStreamThread) {
        SLOG(INFO) << "Task Thread Exit.....";
        if (mStreamThread->joinable()) {
            SLOG(INFO) << "Task Start Before Thread Exiting...";
            mStreamThread->join();
        }
        mStreamThread = nullptr;
    }

    mStop = false;
    mStreamThread = std::make_shared<std::thread>(&StreamPuller::stream_pull_process, this);

    SLOG(INFO) << "Task Started.";
}

bool StreamPuller::isNeedRetry() {
    bool res = retryTimes > retriedTimes;
    if(retryTimes <= 0)
        res = true;
    retriedTimes++;
    SLOG(INFO) << "Task " << mTaskId << " retry: " << retriedTimes << "/" <<retryTimes;
    return res;
}

void StreamPuller::stream_pull_process() {
    mTaskStatus = ThreadStatus::RUNNING;
    bool bOnline = isOnlineUrl();

#ifdef USE_OPENCV
    mCap->open(mURL);
#endif
    if (!mCap->isOpened()) {
        SLOG(WARNING) << "Open URL Failed!!!";
        if (bOnline) {
            mTaskStatus = ThreadStatus::READ_NG;
        } else {
            mTaskStatus = ThreadStatus::STOP_EXCEPTION;
        }
        mErrorHandle();
        if (mOnError) {
            mOnError(mTaskId, mURL);
        }
        return;
    }

#ifdef USE_OPENCV
    int width = (int) mCap->get(cv::CAP_PROP_FRAME_WIDTH);
    int height = (int) mCap->get(cv::CAP_PROP_FRAME_HEIGHT);
    double streamFps = mCap->get(cv::CAP_PROP_FPS);
#else
    int width = mCap->GetWidth();
    int height = mCap->GetHeight();
    double streamFps = mCap->GetFPS();
#endif
    SLOG(INFO) << "Open URL Succeed. Width=" << width << ", Height=" << height << ", FPS=" << streamFps;
    // 根据设定fps进行跳帧解码
    double skip = 0;      // 每解码一帧后需要跳帧的数目
    mUsedFps = this->mFps;// 实际取帧FPS
    // 当设定fps小于0或者大于视频流fps时, 设定值无效, 不进行跳帧处理.
    if (this->mFps > 0 && this->mFps < streamFps) {
        skip = streamFps / this->mFps - 1;
    } else {
        mUsedFps = streamFps;
    }

    double needSkip = 0;         // 需要跳帧数
    uint64_t decodedFrameNum = 0;// 解码帧编号
    uint64_t frameNum = 0;
    cv::Mat frame;
    while (!mStop) {
#ifdef USE_OPENCV
        if (!mCap->read(frame)) {
#else
        Status sts = mCap->Read(frame);
        if (!sts.isOk()) {
            SLOG(WARNING) << sts.message();
#endif
            if (!bOnline) {
                SLOG(INFO) << "Read to end of file!";
            } else {
                SLOG(WARNING) << "Read Frame From URL Failed!!";
            }
            break;
        }

        decodedFrameNum += 1;
        frameNum++;
        retriedTimes = 0;
        SLOG(DEBUG) << "Get Frame: " << decodedFrameNum;
        on_frame(mUsedFps, frame, decodedFrameNum);

        if (skip > 0) {
            needSkip += skip;
            // 跳帧数目需为整数. 当需要跳帧数大于1时进行跳帧处理
            int tmpSkip = cvFloor(needSkip);
            int tmpSkipCnt = 0;
            // 读取包但不解码
            while (tmpSkipCnt < tmpSkip && !mStop) {
#ifdef USE_OPENCV
                if (!mCap->grab()) {
#else
                sts = mCap->Grab();
                if (!sts.isOk()) {
                    SLOG(WARNING) << sts.message();
#endif
                    if (!bOnline) {
                        SLOG(INFO) << "Read to end of file!";
                    } else {
                        SLOG(WARNING) << "Grab Packet From URL Failed!!";
                    }
                    break;
                }
                tmpSkipCnt++;
                frameNum++;
            }
            if (tmpSkipCnt != tmpSkip) {
                // 当跳帧数与预期跳帧数目不一致时, 表明读包错误或者用户停止
                break;
            } else {
                needSkip -= tmpSkip;
                SLOG(DEBUG) << "Skipped Frames: " << tmpSkip;
            }
        }
    }

    if (!mStop) {
        if (bOnline) {
            // 实时流
            mTaskStatus = ThreadStatus::READ_NG;
        } else {
            // 本地文件
            if (frameNum > 0) {
                mTaskStatus = ThreadStatus::STOP_EOF;
            } else {
                mTaskStatus = ThreadStatus::STOP_EXCEPTION;
            }
        }
        mErrorHandle();
        if (mOnError) {
            mOnError(mTaskId, mURL);
        }
    } else {
        mTaskStatus = ThreadStatus::STOP;
    }

#ifdef USE_OPENCV
    mCap->release();
#endif
}

StreamPuller::ThreadStatus StreamPuller::GetTaskThreadStatus() {
    return mTaskStatus;
}

std::string StreamPuller::GetTaskThreadStatusString(StreamPuller::ThreadStatus ts) {
    std::string status;
    switch (ts) {
        case StreamPuller::ThreadStatus::RUNNING:
            status = "Running";
            break;
        case StreamPuller::ThreadStatus::STOP:
            status = "Stopped";
            break;
        case StreamPuller::ThreadStatus::STOP_EOF:
            status = "Stopped(EOF)";
            break;
        case StreamPuller::ThreadStatus::STOP_EXCEPTION:
            status = "Stopped(EXCEPTION)";
            break;
        case StreamPuller::ThreadStatus::READ_NG:
            status = "Retrying";
            break;
        default:
            break;
    }
    return status;
}

void StreamPuller::setOnError(const OnStreamErrorCb &onError) {
    mOnError = onError;
}

Status StreamPuller::InitImageTransportNetwork() {
    if (mCap == nullptr) {
#ifdef USE_OPENCV
        mCap = std::make_shared<cv::VideoCapture>();
#else
        mCap = std::make_shared<VideoReader>(mURL);
#endif
    }

    if (mNetwork == nullptr) {
        mNetwork = std::make_shared<ZMQNetwork>();
    }

    //    std::string address = Config::GetInstance().GetImageZmqServerAddr();
    Status s = mNetwork->SetSocket(ZMQ_REQ, mServerAddress);
    if (!s.isOk()) {
        SLOG(INFO) << s.message();
        SLOG(ERROR) << s.message();
    }
    return s;
}

Status StreamPuller::on_frame(double fps, const cv::Mat &image, uint64_t id) {
    ImageMessage msg{
            mTaskId, image.cols, image.rows, image.type(), id, image};
    mNetwork->SendRequest(&msg, (int) (1000 * 0.8 / fps));
    return Status::OK();
}

Status StreamPuller::mErrorHandle() {
    ImageMessage msg{
        mTaskId, -1, -1, 0, 0, cv::Mat()
    };
    mNetwork->SendRequest(&msg, 1000);
    return Status::OK();
}

bool StreamPuller::isOnlineUrl() {
    std::vector<std::string> onlineTag{
            "udp://", "tcp://", "rtp://", "rtsp://", "rtmp://", "http://"};
    for (const auto &tag : onlineTag) {
        if (StringHelp::StartWith(mURL, tag)) {
            return true;
        }
    }
    return false;
}

std::string StreamPuller::AvError2Str(int err_num) {
    char buf[AV_ERROR_MAX_STRING_SIZE] = {0};
    return std::string(av_make_error_string(buf, AV_ERROR_MAX_STRING_SIZE, err_num));
}

int StreamPuller::DefaultAVInterruptCallBack(void *ptr) {
    auto *pCustomBase = static_cast<StreamPuller *>(ptr);
    int timeout = 3;
    if (av_gettime() - pCustomBase->lastReadPacketTime > timeout * 1000 * 1000) {
        return 1;
    }
    return 0;
}

void StreamPuller::Init() {
#if LIBAVCODEC_VERSION_INT < AV_VERSION_INT(58, 9, 100)
    av_register_all();
#endif
    avformat_network_init();
}

TaskInfo StreamPuller::GetTaskInfo() {
    return TaskInfo{mTaskId, GetTaskThreadStatusString(mTaskStatus), (mTaskStatus == ThreadStatus::RUNNING) ? (int)mUsedFps: mFps};
}

StreamPuller::~StreamPuller() = default;
